package com.atguigu.flink.wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import static org.apache.flink.api.common.typeinfo.Types.*;

/**
 * Created by Smexy on 2022/11/19
 *
 *   flink是一个分布式的计算框架。允许一个算子，并行运算。
 *          并行度越大，针对海量数据时，运行效率高！
 *
 *       并行度如何设置:
 *                  ①全局设置:  启动flink集群时，通过配置文件  parallelism.default: 1
 *                              很少用。 会影响这个集群上所有Job的并行度
 *                  ②提交Job时，也可以指定单个Job的全局并行度
 *                              在代码里:
 *                                           env.setParallelism(1);
 *                              客户端:    webUI 直接写
 *                                      命令行:  -p
 *
 *                 ③每个算子可以单独设置并行度
 *                              在算子的后面.setParallelism(n)
 *
 *                优先级:  ③  >  ②  > ①
 *
 *       某些算子不允许并行运算，不能去设置超过1的并行度！举例:  env.socketTextStream
 *
 *      ----------------------------
 *          flink的算子分为两种:
 *                      两个算子分发规则:  keyBy。 从一个Stream得到另一个Stream，无法调用 .setParallelism(4)
 *                      计算:   只有计算的算子可以设置并行度.
 *                                  计算的算子，返回值都是 XXXOperator，才能设置并行度
 */
public class Demo9_SetParilisim
{
    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",3333);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);

        
        env.setParallelism(1);
        DataStreamSource<String> ds = env.socketTextStream("hadoop103", 8888).setParallelism(1);

        ds.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (value, out) -> {
            String[] words = value.split(" ");
            for (String word : words) {
                out.collect(Tuple2.of(word,1));
            }
        })
          .setParallelism(4)
          .returns(TUPLE(STRING,INT))
          .keyBy( t ->   t.f0)
          .sum(1).setParallelism(2)
          .print().setParallelism(2);


        env.execute();

    }
}
